#if defined(DA_LSM)

#include <string>
#include "buffer.h"

namespace leveldb {
    // On success return the item, or return NULL
    struct bitem* Buffer::Prev(struct bitem* current) {
	struct bitem *prev = NULL;
	struct rb_node *pos;
	pos = rb_prev(&current->bnode);
	if(pos) {
	    prev = rb_entry(pos, struct bitem, bnode);
	}

	return prev;
    }

    // On success return the item, or return NULL
    struct bitem* Buffer::Next(struct bitem* current) {
	struct bitem *next = NULL;
	struct rb_node *pos;
	pos = rb_next(&current->bnode);
	if(pos) {
	    next = rb_entry(pos, struct bitem, bnode);
	}
	return next;
    }

    // On success return 0, on failure return -1 
    int Buffer::Insert(const Slice& key, const Slice& value ) {
	struct rb_node *parent = NULL;
	struct bitem *tmp;
	int ret;

        // alloc and initialize one item
	struct bitem *item = (struct bitem*)malloc(sizeof(struct bitem));
	if(!item) {
	    fprintf(stderr, "Error: malloc() failed at %s, %s:%d\n",
		__func__, __FILE__, __LINE__);
	    return -1;

	}
	item->bkey_data = (char *)malloc(key.size());
	if(!item->bkey_data) {
	    fprintf(stderr, "Error: malloc() failed at %s, %s:%d\n",
		__func__, __FILE__, __LINE__);
	    free(item);
	    return -1;
	}
	memcpy(item->bkey_data, key.data(), key.size());
	item->bkey_size = key.size();
	item->bvalue_size = value.size();

	mutex.Lock();	
	LogWrite(&item->bvalue_offset, key, value);
	dirty_ = true;
	// Insert into rbtree, from the leaf node
	struct rb_node **p = &(broot_.rb_node);
	while(*p) {
	    parent = *p;
	    tmp = rb_entry(parent, struct bitem, bnode);
	    ret = internal_comparator_->Compare(key, Slice(tmp->bkey_data, tmp->bkey_size));
	    if(ret < 0) {
		p = &(*p)->rb_left;
	    } else if(ret > 0) {
		p = &(*p)->rb_right;
	    } else {
		// Does this occur?
		printf("Already exist! %s:%d\n", __FILE__, __LINE__);
		return 0;
	    }
	}

	rb_link_node(&item->bnode, parent, p);
	rb_insert_color(&item->bnode, &broot_);
	size_ += (key.size() + value.size());
	mutex.Unlock();
	return 0;
    }

    // On success return 0;
    // note: key is InternalKey
    int Buffer::Erase(const Slice& key) {
	struct bitem *tmp;
	struct rb_node *pos;

	mutex.Lock();
	tmp = InternalGet(key);
	if(!tmp) {
	    return 0;
	}
	if(internal_comparator_->Compare(key, Slice(tmp->bkey_data,
						    tmp->bkey_size)) == 0) {
	    pos = &tmp->bnode;
	    rb_erase(pos, &broot_);
	    free(tmp->bkey_data);
	    free(tmp);	    
	}
	size_ -= (tmp->bkey_size + tmp->bvalue_size);
	mutex.Unlock();
	return 0;
    }

    // Delete all the items between start_key and end_key, [start_key, end_key];
    // On success return 0.
    // Note: start_key and end_key are all InternalKey.
    int Buffer::Erase(const Slice& start_key, const Slice& end_key) {
	struct bitem *pos;
	struct bitem *start, *end;

	struct rb_node *tmp;
	int ret;

	mutex.Lock();
	// Find the closest item whose bkey is equal or larger than start_key
	pos = start = InternalGet(start_key);
	// Find the closest item whose bkey is equal or larger than end_key
	end = InternalGet(end_key);

	// For special case:
	// end_key is equal to end->bkey;
	if(end &&
	   internal_comparator_->Compare(end_key, Slice(end->bkey_data,
							end->bkey_size)) == 0) {
	    tmp = rb_next(&end->bnode);
	    if(tmp) {
		end = rb_entry(tmp, struct bitem, bnode);
	    } else {
		end = NULL;
	    }
	}
	
	// delete all the items between start_key and end_key
	// that is, [start, end).
	while(pos != end) {
	    tmp = rb_next(&pos->bnode);
	    rb_erase(&pos->bnode, &broot_);
	    free(pos->bkey_data);
	    free(pos);
	    size_ -= (pos->bkey_size + pos->bvalue_size);
	    if(tmp) {
		pos = rb_entry(tmp, struct bitem, bnode);
	    } else {
		pos = NULL;
		break;
	    }
	}

	mutex.Unlock();
	return 0;
    }

    // Find the closest item which is equal to or larger than key which is InternalKey.
    // If there are no equal or larger keys, return NULL; otherwise, return struct bitem*.
    // Note: Before calling _get(), the caller has called the mutex.Lock();
    struct bitem* Buffer::InternalGet(const Slice& key) {
	struct rb_node *prev = NULL;
	struct rb_node *node = NULL;
	struct bitem *tmp = NULL;

	node =broot_.rb_node;
	int ret;

	// main loop
	while(node) {
	    // first, go right;
	    tmp = rb_entry(node, struct bitem, bnode);	    
	    ret = internal_comparator_->Compare(key, Slice(tmp->bkey_data, tmp->bkey_size));
	    while(ret > 0) {
		node = node->rb_right;
		if(node) {
		    tmp = rb_entry(node, struct bitem, bnode);
		    ret = internal_comparator_->Compare(key, Slice(tmp->bkey_data, tmp->bkey_size));
		} else {
		    break;
		}
	    }
	    if(node) {
		prev = node;
	    } else {
		break;
	    }

	    // then, go left
	    while(ret < 0) {
		prev = node;
		node = node->rb_left;
		if(node) {
		    tmp = rb_entry(node, struct bitem, bnode);
		    ret = internal_comparator_->Compare(key, Slice(tmp->bkey_data, tmp->bkey_size));
		} else {
		    break;
		}
	    }

	    // exit
	    if(ret == 0) {
		prev = node;
		break;
	    }	    
	}

	if(prev)
	    return rb_entry(prev, struct bitem, bnode);
	else
	    return NULL;
    }

    // On success return Status::Ok(), and assign the value.
    // Otherwise, return Status::NotFound().
    Status Buffer::Get(const Slice& key, std::string *value) {
	struct bitem *tmp;
	int ret;
	Status status = Status::NotFound(Slice());

	ParsedInternalKey parsed_bkey;
	ParsedInternalKey parsed_key;
	
	mutex.Lock();
	tmp = InternalGet(key);
	if(!tmp) {
	    status = Status::NotFound(Slice());
	    goto out;
	}

	// equal?
	ParseInternalKey(key, &parsed_key);
	ParseInternalKey(Slice(tmp->bkey_data, tmp->bkey_size), \
			 &parsed_bkey);
	
	if(internal_comparator_->user_comparator()->\
	   Compare(parsed_key.user_key, parsed_bkey.user_key) == 0) {
	    if(parsed_bkey.type == kTypeValue) {
		char *data = (char *)malloc(tmp->bvalue_size);
		if(!data) {
		    fprintf(stderr, "malloc() failed in %s:%d\n",
			    __FILE__, __LINE__);
		    exit(1);
		}
		LogRead(tmp->bvalue_offset, tmp->bvalue_size, data);
		value->assign(data, tmp->bvalue_size);
		free(data);
		status = Status();
	    } else {
		status = Status::NotFound(Slice());
	    }
	}
	
    out:
	mutex.Unlock();
	return status;
    }

    Status Buffer::Get(const Slice& key,
		       void *arg) {
	struct LogSaver *saver = (struct LogSaver *)arg;
	struct bitem *tmp;
	int ret;
	Status status = Status::OK();

	ParsedInternalKey parsed_bkey;
	ParsedInternalKey parsed_key;
	
	mutex.Lock();
	tmp = InternalGet(key);
	if(!tmp) {
	    goto out;
	}

	// equal?
	ParseInternalKey(key, &parsed_key);
	ParseInternalKey(Slice(tmp->bkey_data, tmp->bkey_size), \
			 &parsed_bkey);
	
	if(internal_comparator_->user_comparator()->\
	   Compare(parsed_key.user_key, parsed_bkey.user_key) == 0) {
	    // easy for test;
	    // value->assign(tmp->bvalue_data, tmp->bvalue_size);
	    if(parsed_bkey.type == kTypeValue) {
		char *data = (char *)malloc(tmp->bvalue_size);
		if(!data) {
		    fprintf(stderr, "malloc() failed in %s:%d\n",
			    __FILE__, __LINE__);
		    exit(1);
		}
		LogRead(tmp->bvalue_offset, tmp->bvalue_size, data);
		saver->value->assign(data, tmp->bvalue_size);
		free(data);
		saver->state = Found;
	    } else {
		saver->state = Deleted;
	    }
	}
	
    out:
	mutex.Unlock();
	return status;
    }
    
    // If the buffer conains the key, return true; Or, return false.
    // No matter the key type, even is kTypeDeletion.
    bool Buffer::Contains(const Slice& key) {
	struct bitem *tmp;
	int ret;
	bool status = false;

	ParsedInternalKey parsed_bkey;
	ParsedInternalKey parsed_key;
	
	mutex.Lock();
	tmp = InternalGet(key);
	if(!tmp) {
	    status = false;
	    goto out;
	}

	ParseInternalKey(key, &parsed_key);
	ParseInternalKey(Slice(tmp->bkey_data, tmp->bkey_size), \
			 &parsed_bkey);
	
	// Equal no matter the value type.
	if(internal_comparator_->user_comparator()->\
	   Compare(parsed_key.user_key, parsed_bkey.user_key) == 0) {
	    status = true;
	}
	
    out:
	mutex.Unlock();
	return status;
    }

    uint64_t Buffer::Size() {
	return size_;
    }

    int Buffer::Sync() {
	// make sure that all the buf data is written to disk
	if(dirty_ == false) {
	    return 0;
	}
	
	int ret;
	if(wbuf_.offset > last_write_offset_) {
	    // something wrong, lost some data or forget update
	    fprintf(stderr, "error in %s:%d, some data lost.\n",
		    __FILE__, __LINE__);
	    exit(1);
	} else {
	    ret = pwrite(fd_,
			 &wbuf_.data[last_write_offset_ - wbuf_.offset],
			 write_offset_ - last_write_offset_,
			 last_write_offset_);
	    if(ret != write_offset_ - last_write_offset_) {
		fprintf(stderr, "write() not complted in %s:%d\n",
			__FILE__, __LINE__);
		exit(1);
	    }
	}

	// update last_write_offset_
	last_write_offset_ = write_offset_;
	ret = fsync(fd_);
	if(ret != 0) {
	    fprintf(stderr, "fsync() not complted in %s:%d\n",
		    __FILE__, __LINE__);
	    exit(1);
	}
	dirty_ = false;
	return 0;
    }

    int Buffer::LogRead(uint64_t offset, int size, char *data) {
	// use offset and size for read
	char *pos = data;
	int ret;
	int count = size;
	while(count > 0 && offset < wbuf_.offset) {
	    if(offset + count < wbuf_.offset) {
		ret = pread(fd_, pos, count, offset);
	    } else {
		ret = pread(fd_, pos, wbuf_.offset - offset, offset);
	    }

	    if(ret == -1) {
		fprintf(stderr, "pread error in %s:%d\n",
			__FILE__, __LINE__);
		exit(1);
	    }
	    offset += ret;
	    count -= ret;
	    pos += ret; 
	}

	if(count > 0) {
	    if(offset + count > wbuf_.offset + wbuf_.size - wbuf_.available) {
		fprintf(stderr, "pread error in %s:%d\n",
			__FILE__, __LINE__);
		exit(1);
	    }
	    memcpy(pos, &wbuf_.data[offset - wbuf_.offset], count);
	}
	return 0;
    }

    // log entry format:
    // key_size+value_size+key_data+value_data
    // 4+4+key_size+value_size
    int Buffer::LogWrite(uint64_t *bvalue_offset,
			 const Slice& key,
			 const Slice& value) {
	int size = 4 + 4 + key.size() + value.size();
	// need to avoid memory copy and memory allocation
	char *data = (char *)malloc(size);
	if(!data) {
	    fprintf(stderr, "malloc() failed in %s:%d\n",
		    __FILE__, __LINE__);
	    exit(1);
	}
	char *pos = data;
	int key_size = key.size();
	int value_size = value.size();
	memcpy(pos, &key_size, 4);
	pos += 4;
	memcpy(pos, &value_size, 4);
	pos += 4;
	memcpy(pos, key.data(), key_size);
	pos += key_size;
	memcpy(pos, value.data(), value_size);

	*bvalue_offset = write_offset_ + 4 + 4 + key_size;

	pos = data;
	int count = size;
	while(count > 0) {
	    if(wbuf_.available > count) {
		memcpy(&wbuf_.data[wbuf_.size - wbuf_.available],
		       pos,
		       count);
		wbuf_.available -= count;
		write_offset_ += count;
		count = 0;
	    } else {
		memcpy(&wbuf_.data[wbuf_.size - wbuf_.available],
		       pos,
		       wbuf_.available);
		write_offset_ += wbuf_.available;
		count -= wbuf_.available;

		// write buffer log file
		int ret = pwrite(fd_,
				 &wbuf_.data[last_write_offset_ - wbuf_.offset],
				 wbuf_.size - last_write_offset_ + wbuf_.offset,
				 last_write_offset_);
		if(ret != wbuf_.size - last_write_offset_ + wbuf_.offset) {
		    fprintf(stderr, "pwrite() error in %s:%d\n",
			    __FILE__, __LINE__);
		    exit(1);		    
		}
		last_write_offset_ = wbuf_.offset = write_offset_;
		wbuf_.available = wbuf_.size;
	    }
	}
	free(data);
	return 0;
    }

    // When there is Buffer::Biter, cann't change the buffer; or be corrupted.
    class Buffer::BIter : public Iterator {
    public:
	BIter(Buffer *buffer, const InternalKeyComparator *comparator,	\
	      Slice& start_key, Slice& end_key) {
	    buffer_ = buffer;
	    internal_comparator_ = comparator;
	    start_key_.assign(start_key.data(), start_key.size());
	    end_key_.assign(end_key.data(), end_key.size());
	    current_ = NULL;
	    data_ = (char *)malloc(MAX_VALUE_SIZE);
	    if(!data_) {
		fprintf(stderr, "malloc() failed in %s:%d\n",
			__FILE__, __LINE__);
		exit(1);
	    }
	}

	~BIter() {
	    free(data_);
	}

	void AssignValue(struct bitem *current) {
	    if(current_->bvalue_size > MAX_VALUE_SIZE) {
		fprintf(stderr, "Error, buffer size is too small in %s:%d\n",
			__FILE__, __LINE__);
		exit(1);
	    }
	    buffer_->LogRead(current_->bvalue_offset,
			     current_->bvalue_size,
			     data_);
	    value_ = Slice(data_, current_->bvalue_size);
	}
	
	virtual bool Valid() const {
	    if(!current_) {
		return false;
	    }
	    if(internal_comparator_->Compare(key_, start_key_) < 0) {
		return false;
	    }
	    if(internal_comparator_->Compare(key_, end_key_) > 0) {
		return false;
	    }
	    return true;
	}

	void SeekToFirst() {
	    current_ = buffer_->InternalGet(start_key_);
	    if(current_) {
		key_ = Slice(current_->bkey_data, current_->bkey_size);
	    }
	}

	void SeekToLast() {
	    current_ = buffer_->InternalGet(end_key_);
	    if(internal_comparator_->Compare(Slice(end_key_.c_str(),
						      end_key_.size()),
						Slice(current_->bkey_data,
						      current_->bkey_size)) < 0) {
		current_ = buffer_->Prev(current_);
	    }
	    if(current_) {
		key_ = Slice(current_->bkey_data, current_->bkey_size);
		AssignValue(current_);
	    }
	}

	void Seek(const Slice& target) {
	    current_ = buffer_->InternalGet(target);
	    if(current_) {
		key_ = Slice(current_->bkey_data, current_->bkey_size);
		AssignValue(current_);
	    }
	}

	void Next() {
	    current_ = buffer_->Next(current_);
	    if(current_) {
		key_ = Slice(current_->bkey_data, current_->bkey_size);
		AssignValue(current_);
	    }
	}

	void Prev() {
	    current_ = buffer_->Prev(current_);
	    if(current_) {
		key_ = Slice(current_->bkey_data, current_->bkey_size);
		AssignValue(current_);
	    }
	}

	virtual Slice key() const {
	    return key_;
	}

	virtual Slice value() const {
	    return value_;
	}

	virtual Status status() const {
	    return status_;
	}
	
    private:
	std::string start_key_;
	std::string end_key_;
	struct bitem *current_;
	Slice key_;    // Slice(current_->bkey_data, current_->bkey_size)
	Slice value_;  // Slice(current_->bvalue_data, current_->bvalue_size)
	Status status_;
	const InternalKeyComparator *internal_comparator_;
	Buffer *buffer_;

	char *data_;
    };

    // Create an iterator for the buffer, which is valid in [start_key, end_key].
    Iterator* Buffer::NewIterator(Slice& start_key, Slice& end_key) {
	return new BIter(this, internal_comparator_, start_key, end_key);
    }
}
#endif
